"""
Case Type   : Ustore_Standby_Logicreplication
Case Name   : with 语句 insert into逻辑复制
Create At   : 2022/03/03
Owner       : zhao-pengcheng-g
Description :
    1.备份postgresql.conf,pg_hba.conf文件
    2.修改ustore参数
    3.创建数据库，创建用户，修改pg_hba文件,创建表
    4.修改逻辑复制相关参数
    5.创建表，创建逻辑复制槽
    6.开启逻辑复制槽
    7.DML操作
    8.比对数据
    9.停止逻辑复制槽
    10.删除逻辑复制槽
    11.资源清理，恢复参数
Expect      :
    1.备份成功
    2.修改ustore参数成功
    3.创建成功，修改pg_hba文件成功
    4.修改逻辑复制相关参数成功
    5.创建表，创建逻辑复制槽成功
    6.开启逻辑复制槽成功
    7.DML操作执行成功
    8.比对数据成功
    9.停止逻辑复制槽成功
    10.删除逻辑复制槽成功
    11.资源清理成功，恢复参数成功
History     :
"""
import os
import time
import unittest
from yat.test import Node
from yat.test import macro
from testcase.utils.Logger import Logger
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.ComThread import ComThread

Primary_SH = CommonSH('PrimaryDbUser')

@unittest.skipIf(2 != Primary_SH.get_node_num(),'非1+1环境不执行')
class StandbyLogicReplication(unittest.TestCase):
    def setUp(self):
        self.logger = Logger()
        self.logger.info(f"-----{os.path.basename(__file__)} start-----")
        self.primary_node = Node('PrimaryDbUser')
        self.primary_root = Node('PrimaryRoot')
        self.standby_node = Node('Standby1DbUser')
        self.standby_root = Node('Standby1Root')
        self.sh_primy = CommonSH('PrimaryDbUser')
        self.sh_standby = CommonSH('Standby1DbUser')
        self.constant = Constant()
        self.common = Common()
        self.dir_replication = os.path.join(macro.DB_BACKUP_PATH,
                                            'dir_replication_0002')
        self.postgresql_path = os.path.join(macro.DB_INSTANCE_PATH,
                                            macro.DB_PG_CONFIG_NAME)
        self.postgresql_bak_path = os.path.join(self.dir_replication,
                                                'postgresql_bak.conf')
        self.pg_hba_path = os.path.join(macro.DB_INSTANCE_PATH,
                                        macro.PG_HBA_FILE_NAME)
        self.pg_hba_bak_path = os.path.join(self.dir_replication,
                                            'pg_hba_bak.conf')
        self.db_name = 'db_replication_0002'
        self.user_name = 'u_logic_replication_0002'
        self.table_name = 't_standby_logic_replication_0002'
        self.original_table = 'standby_logic_support_type_no_serial_tab'
        self.slot_name_1 = 'slot_standby_logic_replication_0002_1'
        self.slot_name_2 = 'slot_standby_logic_replication_0002_2'
        self.log_file_1 = os.path.join(self.dir_replication,
                                       'log_standby_logic_replication_0002_1')
        self.log_file_2 = os.path.join(self.dir_replication,
                                       'log_standby_logic_replication_0002_2')

        text = "-----step1:备份postgresql.conf,pg_hba.conf;expect:成功-----"
        self.logger.info(text)
        cp_cmd = f'''rm -rf {self.dir_replication} && 
                    mkdir {self.dir_replication} &&
                    cp {self.postgresql_path} {self.postgresql_bak_path} && 
                    cp {self.pg_hba_path} {self.pg_hba_bak_path} && echo OK'''
        cp_pri_msg = self.primary_node.sh(cp_cmd).result()
        self.logger.info(cp_pri_msg)
        self.assertEqual("OK", cp_pri_msg, "执行失败" + text)
        cp_sta_msg = self.standby_node.sh(cp_cmd).result()
        self.logger.info(cp_sta_msg)
        self.assertEqual("OK", cp_sta_msg, "执行失败" + text)

        text = "-----step2:修改ustore参数;expect:成功-----"
        self.logger.info(text)
        self.config_ustore = 'enable_default_ustore_table'
        self.show_ustore_cmd = f'show {self.config_ustore};'
        check_res = self.sh_primy.execut_db_sql(self.show_ustore_cmd)
        self.logger.info(check_res)
        self.ustore_default = check_res.splitlines()[-2].strip()
        if 'on' != self.ustore_default:
            guc_result = self.sh_primy.execute_gsguc('reload',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'{self.config_ustore}=on')
            self.assertTrue(guc_result, '执行失败' + text)
        show_msg = self.sh_primy.execut_db_sql(self.show_ustore_cmd)
        self.logger.info(show_msg)
        self.common.equal_sql_mdg(show_msg, self.config_ustore, 'on',
                                  '(1 row)', flag='1')

        text = "-----step3:创建数据库，创建用户，修改pg_hba文件,创建表;expect:成功-----"
        self.logger.info(text)
        text = "-----step3.1:创建数据库,创建用户,修改pg_hba文件;expect:成功-----"
        self.logger.info(text)
        create_cmd = f'''drop database if exists {self.db_name};
                    create database {self.db_name} template template0 \
                    encoding 'UTF8' lc_ctype 'en_US.UTF-8' \
                    lc_collate 'en_US.UTF-8';
                    drop user if exists {self.user_name};
                    create user {self.user_name} sysadmin \
                    password '{macro.COMMON_PASSWD}';'''
        create_msg = self.sh_primy.execut_db_sql(create_cmd,
                                                 dbname='postgres')
        self.logger.info(create_msg)
        self.assertIn(self.constant.CREATE_DATABASE_SUCCESS and
                      self.constant.CREATE_ROLE_SUCCESS_MSG,
                      create_msg, '执行失败' + text)
        param_1 = f'local replication all trust'
        param_2 = f'host replication {self.user_name} 127.0.0.1/32 sha256'
        param_3 = f'host all {self.user_name} ::1/128 sha256'
        result_1 = self.sh_primy.execute_gsguc('reload',
                                               self.constant.GSGUC_SUCCESS_MSG,
                                               '', pghba_param=param_1)
        result_2 = self.sh_primy.execute_gsguc('reload',
                                               self.constant.GSGUC_SUCCESS_MSG,
                                               '', pghba_param=param_2)
        result_3 = self.sh_primy.execute_gsguc('reload',
                                               self.constant.GSGUC_SUCCESS_MSG,
                                               '', pghba_param=param_3)
        self.assertTrue(result_1 and result_2 and result_3, '执行失败' + text)

        text = "-----step:3.2创建表;expect:成功-----"
        self.logger.info(text)
        self.common.scp_file(self.primary_node, 'table_and_data.sql',
                             self.dir_replication)
        self.sql_file_path = os.path.join(self.dir_replication,
                                          'table_and_data.sql')
        execute_cmd = f"source {macro.DB_ENV_PATH};" \
            f"gsql -d {self.db_name} " \
            f"-p {self.primary_node.db_port} " \
            f"-f {self.sql_file_path}"
        execute_msg = self.primary_node.sh(execute_cmd).result()
        self.logger.info(execute_msg)
        self.assertNotIn("ERROR", execute_msg, '执行失败' + text)

        text = "-----step4:修改逻辑复制相关参数;expect:成功-----"
        self.logger.info(text)
        self.config_slot_log = 'enable_slot_log'
        self.show_slot_log_cmd = f'show {self.config_slot_log};'
        check_res = self.sh_primy.execut_db_sql(self.show_slot_log_cmd)
        self.logger.info(check_res)
        self.slot_log_default = check_res.splitlines()[-2].strip()
        if 'on' != self.slot_log_default:
            guc_result = self.sh_primy.execute_gsguc('reload',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'{self.config_slot_log}=on')
            self.assertTrue(guc_result, '执行失败' + text)
        show_msg = self.sh_primy.execut_db_sql(self.show_slot_log_cmd)
        self.logger.info(show_msg)
        self.common.equal_sql_mdg(show_msg, self.config_slot_log,
                                  'on', '(1 row)', flag='1')

        self.config_wal_level = 'wal_level'
        self.show_wal_level_cmd = f'show {self.config_wal_level};'
        check_res = self.sh_primy.execut_db_sql(self.show_wal_level_cmd)
        self.logger.info(check_res)
        self.wal_level_default = check_res.splitlines()[-2].strip()
        if 'logical' != self.wal_level_default:
            guc_result = self.sh_primy.execute_gsguc('set',
                                        self.constant.GSGUC_SUCCESS_MSG,
                                        f'{self.config_wal_level}=logical')
            self.assertTrue(guc_result, '执行失败' + text)
            restart_msg = self.sh_primy.restart_db_cluster()
            self.logger.info(restart_msg)
            self.assertTrue(restart_msg, '重启失败')
        show_msg = self.sh_primy.execut_db_sql(self.show_wal_level_cmd)
        self.logger.info(show_msg)
        self.common.equal_sql_mdg(show_msg, self.config_wal_level,
                                  'logical', '(1 row)', flag='1')
        status = self.sh_primy.get_db_cluster_status()
        self.assertNotIn("ERROR", status, '执行失败:' + text)
        self.assertIn("Normal", status, '执行失败:' + text)

    def test_standby_logicreplication(self):
        text = "-----step5:创建表，创建逻辑复制槽;expect:成功-----"
        self.logger.info(text)
        create_cmd = f'''drop table if exists {self.table_name};
            create table {self.table_name} (like {self.original_table});
            truncate {self.original_table};
            alter table {self.table_name} replica identity full;
            alter table {self.table_name} \
            add constraint par_p_k_0002 primary key(col_int);
            call insert_to_table(1, 2, '{self.original_table}');'''
        create_msg = self.sh_primy.execut_db_sql(create_cmd,
                                                 dbname=self.db_name)
        self.logger.info(create_msg)
        self.assertTrue(create_msg.count(self.constant.ALTER_TABLE_MSG) == 3,
                        '执行失败:' + text)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS and
                      self.constant.CREATE_TABLE_SUCCESS and
                      self.constant.TRUNCATE_SUCCESS_MSG and
                      "insert_to_table", create_msg, '执行失败:' + text)
        create_slot_1 = f"pg_recvlogical -d {self.db_name} " \
            f"-S {self.slot_name_1} " \
            f"-p {self.primary_node.db_port} " \
            f"-U {self.user_name} " \
            f"-s 2 --create"
        execute_cmd = f'''source {macro.DB_ENV_PATH};
                                  expect <<EOF
                                  set timeout 300
                                  spawn {create_slot_1}
                                  expect "Password:"
                                  send "{macro.COMMON_PASSWD}\\n"
                                  expect eof\n''' + '''EOF'''
        self.logger.info(execute_cmd)
        execute_msg = self.primary_node.sh(execute_cmd).result()
        self.logger.info(execute_msg)
        create_slot_2 = f"pg_recvlogical -d {self.db_name} " \
            f"-S {self.slot_name_2} " \
            f"-p {self.primary_node.db_port} " \
            f"-U {self.user_name} " \
            f"-s 2 --create"
        execute_cmd = f'''source {macro.DB_ENV_PATH}; 
                                  expect <<EOF 
                                  set timeout 300 
                                  spawn {create_slot_2} 
                                  expect "Password:" 
                                  send "{macro.COMMON_PASSWD}\\n" 
                                  expect eof\n''' + '''EOF'''
        self.logger.info(execute_cmd)
        execute_msg = self.primary_node.sh(execute_cmd).result()
        self.logger.info(execute_msg)
        text = "-----验证创建逻辑复制槽成功;expect:成功-----"
        self.logger.info(text)
        sql_cmd = f"select slot_name,plugin,slot_type,active," \
            f"restart_lsn from pg_get_replication_slots();"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname=self.db_name)
        self.logger.info(sql_msg)
        self.assertIn(self.slot_name_1 and self.slot_name_2,
                      sql_msg, '执行失败:' + text)

        text = "-----step6:开启逻辑复制槽;expect:成功-----"
        self.logger.info(text)
        primy_start_cmd = f"pg_recvlogical " \
            f"-o skip-empty-xacts=true -F 1 " \
            f"-d {self.db_name} -S {self.slot_name_1} " \
            f"-p {self.primary_node.db_port} -s 2 " \
            f"-f {self.log_file_1} " \
            f"-U {self.user_name} " \
            f"--start "
        execute_cmd = f'''source {macro.DB_ENV_PATH}; 
                                  expect <<EOF 
                                  set timeout 300 
                                  spawn {primy_start_cmd} 
                                  expect "Password:" 
                                  send "{macro.COMMON_PASSWD}\\n" 
                                  expect eof\n''' + '''EOF'''
        self.logger.info(execute_cmd)
        thread_1 = ComThread(self.common.get_sh_result,
                             args=(self.primary_node, execute_cmd,))
        thread_1.setDaemon(True)
        thread_1.start()
        thread_1.join(10)
        msg_result_1 = thread_1.get_result()
        self.logger.info(msg_result_1)
        text = "-----验证主机开启逻辑复制槽成功;expect:成功-----"
        self.logger.info(text)
        sql_cmd = f"select active from pg_get_replication_slots() " \
            f"where slot_name like '{self.slot_name_1}%';;"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname=self.db_name)
        self.logger.info(sql_msg)
        self.assertEqual('t', sql_msg.splitlines()[-2].strip(),
                         '执行失败:' + text)

        stanby_start_cmd = f"pg_recvlogical " \
            f"-o skip-empty-xacts=true -F 1 " \
            f"-d {self.db_name} -S {self.slot_name_2} " \
            f"-p {self.standby_node.db_port} -s 2 " \
            f"-f {self.log_file_2} " \
            f"-U {self.user_name} " \
            f"--start "
        execute_cmd = f'''source {macro.DB_ENV_PATH}; 
                                expect <<EOF 
                                set timeout 300 
                                spawn {stanby_start_cmd} 
                                expect "Password:" 
                                send "{macro.COMMON_PASSWD}\\n" 
                                expect eof\n''' + '''EOF'''
        self.logger.info(execute_cmd)
        thread_2 = ComThread(self.common.get_sh_result,
                             args=(self.standby_node, execute_cmd,))
        thread_2.setDaemon(True)
        thread_2.start()
        thread_2.join(10)
        msg_result_2 = thread_2.get_result()
        self.logger.info(msg_result_2)
        text = "-----验证备机开启逻辑复制槽成功;expect:成功-----"
        self.logger.info(text)
        sql_cmd = f"select active from pg_get_replication_slots() " \
            f"where slot_name like '{self.slot_name_2}%';;"
        sql_msg = self.sh_standby.execut_db_sql(sql_cmd, dbname=self.db_name)
        self.logger.info(sql_msg)
        self.assertEqual('t', sql_msg.splitlines()[-2].strip(),
                         '执行失败:' + text)

        text = "-----step7:DML操作;expect:成功-----"
        self.logger.info(text)
        sql_cmd = f'''with t1 as (select * from {self.original_table}) \
            insert into {self.table_name} select * from t1;
            with t1 as (select * from {self.original_table}) \
            insert into {self.table_name}(col_int) select col_int+10 from t1;'''
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname=self.db_name)
        self.logger.info(sql_msg)
        self.assertNotIn("ERROR", sql_msg, "执行失败" + text)

        text = "-----step8:比对数据;expect:成功-----"
        self.logger.info(text)
        shell_cmd_1 = f"sleep 10 && cat {self.log_file_1} "
        self.logger.info(shell_cmd_1)
        shell_msg_1 = self.primary_node.sh(shell_cmd_1).result()
        self.logger.info(shell_msg_1)
        self.assertNotIn("ERROR", shell_msg_1, "执行失败" + text)
        shell_cmd_2 = f"sleep 10 && cat {self.log_file_2} "
        self.logger.info(shell_cmd_2)
        shell_msg_2 = self.standby_node.sh(shell_cmd_2).result()
        self.logger.info(shell_msg_2)
        self.assertNotIn("ERROR", shell_msg_2, "执行失败" + text)
        self.assertEqual(shell_msg_1, shell_msg_2, "执行失败" + text)
        time.sleep(5)

        text = "-----step9:停止逻辑复制槽;expect:成功-----"
        self.logger.info(text)
        stop_cmd = "ps -ef | grep  pg_recvlogical | grep -v grep | " \
                   "awk '{{print $2}}' | xargs kill -9"
        self.logger.info(stop_cmd)
        primy_result = self.primary_node.sh(stop_cmd).result()
        self.logger.info(primy_result)
        standy_result = self.standby_node.sh(stop_cmd).result()
        self.logger.info(standy_result)
        time.sleep(5)

    def tearDown(self):
        text = "-----step10:删除逻辑复制槽;expect:成功-----"
        self.logger.info(text)
        drop_slot_cmd_1 = f"source {macro.DB_ENV_PATH};" \
            f"pg_recvlogical -d {self.db_name} " \
            f"-S {self.slot_name_1} " \
            f"-p {self.primary_node.db_port} " \
            f"-s 2 --drop"
        drop_slot_msg_1 = self.primary_node.sh(drop_slot_cmd_1).result()
        self.logger.info(drop_slot_msg_1)
        drop_slot_cmd_2 = f"source {macro.DB_ENV_PATH};" \
            f"pg_recvlogical -d {self.db_name} " \
            f"-S {self.slot_name_2} " \
            f"-p {self.primary_node.db_port} " \
            f"-s 2 --drop"
        drop_slot_msg_2 = self.primary_node.sh(drop_slot_cmd_2).result()
        self.logger.info(drop_slot_msg_2)
        self.logger.info("-----验证是否删除成功;expect:成功-----")
        sql_cmd = f"select slot_name,plugin,slot_type,active," \
            f"restart_lsn from pg_get_replication_slots();"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname=self.db_name)
        self.logger.info(sql_msg)
        
        text = "-----step11:资源清理，恢复参数，expect:成功-----"
        self.logger.info(text)
        text_1 = "-----step11.1:删除数据库，删除用户，expect:成功-----"
        self.logger.info(text_1)
        drop_cmd = f'''drop database {self.db_name};
            drop user {self.user_name};'''
        drop_msg = self.sh_primy.execut_db_sql(drop_cmd, dbname='postgres')
        self.logger.info(drop_msg)

        text_2 = "-----step11.2:恢复postgresql.conf,pg_hba.conf文件，" \
                 "expect:成功-----"
        self.logger.info(text_2)
        restore_cmd_1 = f'''
            if [[ -e {self.postgresql_path} && -e {self.postgresql_bak_path} ]]; 
            then rm -rf {self.postgresql_path}; 
            mv {self.postgresql_bak_path} {self.postgresql_path}; 
            echo OK;
            else echo not exits; fi'''
        self.logger.info(restore_cmd_1)
        restore_primary_1 = self.primary_node.sh(restore_cmd_1).result()
        restore_standby_1 = self.standby_node.sh(restore_cmd_1).result()
        self.logger.info(restore_primary_1)
        self.logger.info(restore_standby_1)
        restore_cmd_2 = f'''
            if [[ -e {self.pg_hba_path} && -e {self.pg_hba_bak_path} ]];
            then rm -rf {self.pg_hba_path};  
            mv {self.pg_hba_bak_path} {self.pg_hba_path};
            echo OK;
            else echo not exits; fi'''
        self.logger.info(restore_cmd_2)
        restore_primary_2 = self.primary_node.sh(restore_cmd_2).result()
        restore_standby_2 = self.standby_node.sh(restore_cmd_2).result()
        self.logger.info(restore_primary_2)
        self.logger.info(restore_standby_2)

        text_3 = "-----step11.3:删除文件，expect:成功-----"
        self.logger.info(text_3)
        del_cmd = f'rm -rf {self.dir_replication};'
        self.logger.info(del_cmd)
        self.primary_root.sh(del_cmd)
        self.standby_root.sh(del_cmd)
        file_cmd = f'''if [ -d {self.dir_replication} ]; 
            then echo "exists"; else echo "not exists"; fi'''
        primy_file_msg = self.primary_node.sh(file_cmd).result()
        standy_file_msg = self.standby_node.sh(file_cmd).result()
        self.logger.info(primy_file_msg)
        self.logger.info(standy_file_msg)

        self.assertNotIn(self.slot_name_1 and self.slot_name_2,
                         sql_msg, '执行失败:' + text)
        self.assertIn(self.constant.DROP_DATABASE_SUCCESS and
                      self.constant.DROP_ROLE_SUCCESS_MSG,
                      drop_msg, '执行失败' + text_1)
        self.assertEqual('OK', restore_primary_1, '执行失败' + text_2)
        self.assertEqual('OK', restore_standby_1, '执行失败' + text_2)
        self.assertEqual('OK', restore_primary_2, '执行失败' + text_2)
        self.assertEqual('OK', restore_standby_2, '执行失败' + text_2)
        self.assertEqual("not exists", primy_file_msg, '执行失败' + text_3)
        self.assertEqual("not exists", standy_file_msg, '执行失败' + text_3)
        self.logger.info(f"-----{os.path.basename(__file__)} end-----")
